package org.smartcti.freeswitch.inbound.internal;


import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.smartcti.freeswitch.inbound.InboundConnection;
import org.smartcti.freeswitch.inbound.InboundConnectionState;
import org.smartcti.freeswitch.inbound.InboundEventListener;
import org.smartcti.freeswitch.inbound.InboundException;
import org.smartcti.freeswitch.inbound.event.DisconnectEvent;
import org.smartcti.freeswitch.inbound.event.InboundEvent;
import org.smartcti.freeswitch.inbound.response.CommandResponse;

import org.smartcti.freeswitch.util.SocketConnectionFacade;
import org.smartcti.freeswitch.util.internal.MessageFormatter;
import org.smartcti.freeswitch.util.internal.SocketConnectionFacadeImpl;

public class InboundConnectionImpl implements InboundConnection, Dispatcher {
	private static final int RECONNECTION_INTERVAL_1 = 50;
	private static final int RECONNECTION_INTERVAL_2 = 5000;

	private boolean keepAliveAfterAuthenticationFailure = true;

	private final String hostname;
	private final int port;
	private final String password;
	private final List<InboundEventListener> eventListeners;
	private int socketTimeout = Integer.MAX_VALUE;
	private int socketReadTimeout = Integer.MAX_VALUE;
	private boolean ssl = false;
	private InboundReader reader;
	private SocketConnectionFacade socket;
	private Thread readerThread;
	private InboundConnectionState state = InboundConnectionState.INITIAL;
	private static final AtomicLong idCounter = new AtomicLong(0);
	private final AtomicLong reconnectThreadCounter = new AtomicLong(0);
	private final long id;
	private String eventMask;
	private final AtomicLong readerThreadCounter = new AtomicLong(0);
	private final BlockingQueue<CommandResponse> responseQueue = new LinkedBlockingQueue<CommandResponse>();

	private final Logger logger = LogManager
			.getLogger(InboundConnectionImpl.class);

	public InboundConnectionImpl(String hostname, int port, String password) {
		this.id = idCounter.getAndIncrement();
		this.hostname = hostname;
		this.password = password;
		this.port = port;
		eventListeners = new ArrayList<InboundEventListener>();
	}

	public synchronized void addEventListener(InboundEventListener eventListener) {
		eventListeners.add(eventListener);
	}

	public String getHostname() {
		return hostname;
	}

	public String getPassword() {
		return password;
	}

	public int getPort() {
		return port;
	}

	public InboundConnectionState getState() {
		return this.state;
	}

	public synchronized void logoff() throws InboundException {
		// TODO Auto-generated method stub

	}

	public synchronized void removeEventListener(
			InboundEventListener eventListener) {
		eventListeners.remove(eventListener);
	}

	public synchronized CommandResponse sendCommand(String command,
			Object... objects) throws InboundException {
		if (state != InboundConnectionState.CONNECTED) {
			throw new IllegalStateException(
					"Command may only be sent when in state "
							+ "CONNECTED, but connection is in state " + state);
		}
		final String inputCommand = MessageFormatter.arrayFormat(command,
				objects);
		return sendSocket(inputCommand);
	}

	private CommandResponse sendSocket(String command) throws InboundException {
		if (socket == null) {
			throw new IllegalStateException(
					"Command may only be sent when in socket is opened.");
		}
		final ResponseHandlerResult result = new ResponseHandlerResult();
		try {
			synchronized (result) {
				logger.debug("Send socket:" + command);
				socket.write(command + "\n\n");
				socket.flush();
				final CommandResponse response = responseQueue.take();
				result.setResponse(response);
				return result.getResponse();
			}
		} catch (IOException e) {
			e.printStackTrace();
			throw new InboundException(e);
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new InboundException(e);
		}
	}

	public void setSocketReadTimeout(int socketReadTimeout) {
		this.socketReadTimeout = socketReadTimeout;
	}

	public void setSocketTimeout(int socketTimeout) {
		this.socketTimeout = socketTimeout;
	}

	public void setSsl(boolean ssl) {
		this.ssl = ssl;
	}

	public int getSocketTimeout() {
		return socketTimeout;
	}

	public int getSocketReadTimeout() {
		return socketReadTimeout;
	}

	public boolean isSsl() {
		return ssl;
	}

	protected synchronized void connect() throws InboundException {
		logger.info("Connecting to " + hostname + ":" + port);

		if (reader == null) {
			logger.debug("Creating reader for " + hostname + ":" + port);
			reader = new InboundReaderImpl(this);
		}

		logger.debug("Creating socket");
		socket = createSocket();

		logger.debug("Passing socket to reader");
		reader.setSocket(socket);

		if (readerThread == null || !readerThread.isAlive() || reader.isDead()) {
			logger.debug("Creating and starting reader thread");
			readerThread = new Thread(reader);
			readerThread.setName("FreeSwitch-Java Event Socket Connection-"
					+ id + "-Reader-" + readerThreadCounter.getAndIncrement());
			readerThread.setDaemon(true);
			readerThread.start();
		}

		logger.debug("Passing socket to writer");
	}

	protected SocketConnectionFacade createSocket() throws InboundException {
		try {
			return new SocketConnectionFacadeImpl(hostname, port, ssl,
					socketTimeout, socketReadTimeout);
		} catch (IOException e) {
			e.printStackTrace();
			throw new InboundException(e);
		}
	}

	public synchronized void login() throws InboundException {
		login("event plain ALL");
	}

	public synchronized void login(String eventMask) throws InboundException {
		if (socket == null) {
			connect();
		}

		if (state != InboundConnectionState.INITIAL
				&& state != InboundConnectionState.DISCONNECTED) {
			throw new InboundException(
					"Login may only be perfomed when in state "
							+ "INITIAL or DISCONNECTED, but connection is in state "
							+ state);
		}

		state = InboundConnectionState.CONNECTING;
		this.eventMask = eventMask;
		try {
			final CommandResponse authResponse = sendSocket("auth "
					+ this.password);
			if (authResponse.isSuccess()) {
				final CommandResponse eventOpenResponse = sendSocket(eventMask);
				if (eventOpenResponse.isSuccess()) {
					state = InboundConnectionState.CONNECTED;
				}
			}
		} finally {
			if (state != InboundConnectionState.CONNECTED) {
				state = InboundConnectionState.DISCONNECTED;
			}
		}
	}

	public void dispatchEvent(InboundEvent event) {
		// shouldn't happen
		if (event == null) {
			logger
					.error("Unable to dispatch null event. This should never happen. Please file a bug.");
			return;
		}

		logger.debug("Dispatching event:\n" + event.toString());
		if (event instanceof DisconnectEvent) {
			if (state == InboundConnectionState.CONNECTED) {
				state = InboundConnectionState.RECONNECTING;
				Thread reconnectThread = new Thread(new Runnable()
                {
                    public void run()
                    {
                        reconnect();
                    }
                });
				reconnectThread.setName("Asterisk-Java ManagerConnection-" + id + "-Reconnect-"
                        + reconnectThreadCounter.getAndIncrement());
                reconnectThread.setDaemon(true);
                reconnectThread.start();
			}else{
				return;
			}

		}

		synchronized (eventListeners) {
			for (InboundEventListener listener : eventListeners) {
				try {
					listener.onInboundEvent(event);
				} catch (RuntimeException e) {
					logger.warn("Unexpected exception in eventHandler "
							+ listener.getClass().getName(), e);
				}
			}
		}

	}

	public void dispatchResponse(CommandResponse response) {
		if (response != null) {
			try {
				logger.debug("Got reponse: " + response);
				responseQueue.put(response);
			} catch (InterruptedException e) {
				logger.error(e);
				e.printStackTrace();
			}
		} else {
			logger
					.error("Unable to dispatch null response. This should never happen. Please file a bug");
		}
	}

	private static class ResponseHandlerResult implements Serializable {
		/**
		 * Serializable version identifier.
		 */
		private static final long serialVersionUID = 7831097958568769220L;
		private CommandResponse response;

		public ResponseHandlerResult() {
		}

		public CommandResponse getResponse() {
			return this.response;
		}

		public void setResponse(CommandResponse response) {
			this.response = response;
		}
	}

	private void reconnect() {
		int numTries;

		// try to reconnect
		numTries = 0;
		while (state == InboundConnectionState.RECONNECTING) {
			try {
				if (numTries < 10) {
					// try to reconnect quite fast for the firt 10 times
					// this succeeds if the server has just been restarted
					Thread.sleep(RECONNECTION_INTERVAL_1);
				} else {
					// slow down after 10 unsuccessful attempts asuming a
					// shutdown of the server
					Thread.sleep(RECONNECTION_INTERVAL_2);
				}
			} catch (InterruptedException e1) {
				// ignore
			}

			try {
				connect();

				try {
					login(eventMask);
					logger.info("Successfully reconnected.");
					// everything is ok again, so we leave
					// when successful doLogin set the state to CONNECTED so no
					// need to adjust it
					break;
				} catch (InboundException e1) {
					if (keepAliveAfterAuthenticationFailure) {
						logger.error("Unable to log in after reconnect: "
								+ e1.getMessage());
					} else {
						logger.error("Unable to login after reconnect: "
								+ e1.getMessage() + ". Giving up.");
						state = InboundConnectionState.DISCONNECTED;
					}
				}

			} catch (InboundException e) {
				// server seems to be still down, just continue to attempt
				// reconnection
				logger.warn("Exception while trying to reconnect: "
						+ e.getMessage());
			}
			numTries++;
		}
	}

	private void cleanup() {
		disconnect();
		this.readerThread = null;
	}

	protected synchronized void disconnect() {
		if (socket != null) {
			logger.info("Closing socket.");
			try {
				socket.close();
			} catch (IOException ex) {
				logger.warn("Unable to close socket: " + ex.getMessage());
			}
			socket = null;
		}
	}

}
